package sc.love.boot.kafka;


import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.SerializationUtils;
import sc.love.boot.mail.MailTools;

import java.util.Optional;


@Component
public class KafkaConsumer {

    @Autowired
    private MailTools mailTools;


    @KafkaListener(topics = {"gologin"})
    public void consumer1(ConsumerRecord<?, ?> record) {
        System.out.println("kafka start consumer login");
        Optional<?> value = Optional.ofNullable(record.value());
        if (value.isPresent()) {
            String o = (String) value.get();
            mailTools.send("login", "有人登錄:" + o, "login", "yangjl5@asiainfo.com,2428351998@qq.com");
            System.out.println("kafka end consumer----------login " + o);
        }
    }

    @KafkaListener(topics = {"jenkis"})
    public void consumer2(ConsumerRecord<?, ?> record) {
        System.out.println("kafka start consumer jenkis");
        Optional<?> value = Optional.ofNullable(record.value());
        if (value.isPresent()) {
            String o = (String) value.get();
            Message message = JSONObject.parseObject(o, Message.class);
            mailTools.send("jenkis", message.getMsg() + o, "jenkis", "yangjl5@asiainfo.com,wush5@asiainfo.com");
            System.out.println("kafka end consumer---------- jenkis " + o);
        }
    }
}
